package com.hzy.utils;

import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.json.JsonObject;
import io.vertx.redis.client.Redis;
import io.vertx.redis.client.RedisAPI;
import io.vertx.redis.client.RedisOptions;
import io.vertx.redis.client.Response;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Stream;

/**
 * Vert.x Redis 客户端工具类。
 * 封装了 Redis 客户端的初始化和常用异步操作。
 */
public class RedisUtil {

    private static final Logger LOGGER = LoggerFactory.getLogger(RedisUtil.class);

    private static RedisAPI redisAPI;
    private static Redis redisClient;

    /**
     * 初始化 Redis 客户端。
     * 应该在 Vert.x 应用启动时，通常在 MainVerticle 中调用一次。
     *
     * @param vertx  Vertx 实例
     * @param config Redis 配置（从 config.json 中读取）
     * @return 返回一个 Future<Void>，表示 Redis 客户端是否成功初始化和连接。
     */
    public static Future<Void> initialize(Vertx vertx, JsonObject config) {
        if (redisAPI != null) {
            LOGGER.warn("RedisUtil has already been initialized.");
            return Future.succeededFuture();
        }

        JsonObject redisConfig = config.getJsonObject("redis");
        if (redisConfig == null) {
            return Future.failedFuture("Redis configuration not found in appConfig.");
        }

        String redisHost = redisConfig.getString("host", "127.0.0.1");
        int redisPort = redisConfig.getInteger("port", 6379);
        String redisPassword = redisConfig.getString("password", "");
        int poolSize = redisConfig.getInteger("poolSize", 10);

        RedisOptions redisOptions = new RedisOptions()
                .setConnectionString(String.format("redis://%s:%d", redisHost, redisPort))
                .setMaxPoolSize(poolSize) // 设置连接池大小
                .setPoolCleanerInterval(2000); // 连接池清理间隔，毫秒

        if (redisPassword != null && !redisPassword.isEmpty()) {
            redisOptions.setPassword(redisPassword);
        }

        redisClient = Redis.createClient(vertx, redisOptions);

        return redisClient.connect()
                .onSuccess(conn -> {
                    redisAPI = RedisAPI.api(conn);
                    LOGGER.info("RedisUtil initialized and connected to Redis successfully at {}:{}", redisHost, redisPort);
                })
                .onFailure(throwable -> {
                    LOGGER.error("Failed to connect to Redis during initialization: {}", throwable.getMessage(), throwable);
                })
                .mapEmpty(); // 成功时返回 Future<Void>
    }

    /**
     * 关闭 Redis 客户端。
     * 应该在 Vert.x 应用停止时调用。
     *
     * @return 返回一个 Future<Void>，表示 Redis 客户端是否成功关闭。
     */
    public static Future<Void> close() {
        if (redisClient != null) {
            redisClient.close();
        }
        return Future.succeededFuture();
    }


    private static Future<RedisAPI> getRedisAPI() {
        if (redisAPI == null) {
            return Future.failedFuture("RedisUtil has not been initialized. Call initialize() first.");
        }
        return Future.succeededFuture(redisAPI);
    }

    /**
     * 设置键值对。
     * HINT: NX代表不存在时才设置，XX代表存在时才设置，EX/PX设置过期时间
     *
     * @param key   键
     * @param value 值
     * @return Future<Void>
     */
    public static Future<Void> set(String key, String value) {
        return getRedisAPI()
                .compose(api -> api.set(List.of(key, value)))
                .<Void>mapEmpty()
                .onFailure(throwable -> LOGGER.error("Failed to GET key '{}': {}", key, throwable.getMessage(), throwable));
    }

    /**
     * 设置键值对并设置过期时间（秒）。
     *
     * @param key     键
     * @param value   值
     * @param seconds 过期时间（秒）
     * @return Future<Void>
     */
    public static Future<Void> setex(String key, long seconds, String value) {
        Future<RedisAPI> apiFuture = getRedisAPI();
        return apiFuture.compose(api -> api.set(List.of(key, value, "EX", String.valueOf(seconds))))
                .<Void>mapEmpty()
                .onFailure(throwable -> LOGGER.error("Failed to GET key '{}': {}", key, throwable.getMessage(), throwable));
    }

    /**
     * 获取键对应的值。
     *
     * @param key 键
     * @return Future<String>，如果键不存在则返回null
     */
    public static Future<String> get(String key) {
        return getRedisAPI()
                .compose(api -> api.get(key))
                .map(response -> response != null ? response.toString() : null)
                .onFailure(throwable -> LOGGER.error("Failed to GET key '{}': {}", key, throwable.getMessage(), throwable));
    }

    /**
     * 删除一个或多个键。
     *
     * @param keys 要删除的键列表
     * @return Future<Long> 成功删除的键的数量
     */
    public static Future<Long> del(List<String> keys) {
        return getRedisAPI()
                .compose(api -> api.del(keys))
                .map(Response::toLong)
                .onFailure(throwable -> LOGGER.error("Failed to DEL keys '{}': {}", keys, throwable.getMessage(), throwable));
    }

    /**
     * 检查键是否存在。
     *
     * @param key 键
     * @return Future<Boolean>
     */
    public static Future<Boolean> exists(String key) {
        return getRedisAPI()
                .compose(api -> api.exists(List.of(key)))
                .map(response -> response != null && response.toLong() == 1L)
                .onFailure(throwable -> LOGGER.error("Failed to EXISTS key '{}': {}", key, throwable.getMessage(), throwable));
    }

    /**
     * 获取哈希表中指定字段的值。
     *
     * @param key   哈希表的键
     * @param field 字段
     * @return Future<String>，如果键或字段不存在则返回null
     */
    public static Future<String> hget(String key, String field) {
        return getRedisAPI()
                .compose(api -> api.hget(key, field))
                .map(response -> response != null ? response.toString() : null)
                .onFailure(throwable -> LOGGER.error("Failed to HGET key '{}', field '{}': {}", key, field, throwable.getMessage(), throwable));
    }

    /**
     * 设置哈希表中指定字段的值。
     *
     * @param key   哈希表的键
     * @param field 字段
     * @param value 值
     * @return Future<Void>
     */
    public static Future<Void> hset(String key, String field, String value) {
        return getRedisAPI()
                .compose(api -> api.hset(List.of(key, field, value)))
                .<Void>mapEmpty()
                .onFailure(throwable -> LOGGER.error("Failed to HSET key '{}', field '{}': {}", key, field, throwable.getMessage(), throwable));
    }

    /**
     * 设置哈希表中多个字段的值。
     *
     * @param key 哈希表的键
     * @param map 包含字段-值对的 Map
     * @return Future<Void>
     */
    public static Future<Void> hmset(String key, Map<String, String> map) {
        List<String> args = map.entrySet().stream()
                .flatMap(entry -> Stream.of(entry.getKey(), entry.getValue()))
                .toList();
        return getRedisAPI()
                .compose(api -> api.hmset(new ArrayList<>() {{
                    add(key);
                    addAll(args);
                }}))
                .<Void>mapEmpty()
                .onFailure(throwable -> LOGGER.error("Failed to HMSET key '{}': {}", key, throwable.getMessage(), throwable));
    }

    /**
     * 获取哈希表中所有字段和值。
     *
     * @param key 哈希表的键
     * @return Future<JsonObject> 包含所有字段-值对的 JsonObject
     */
    public static Future<JsonObject> hgetall(String key) {
        return getRedisAPI()
                .compose(api -> api.hgetall(key))
                .map(response -> {
                    if (response != null && response.size() > 0) {
                        JsonObject result = new JsonObject();
                        // Redis hgetall 返回的是 key1, value1, key2, value2... 这样的扁平列表
                        for (int i = 0; i < response.size(); i += 2) {
                            String field = response.get(i).toString();
                            String value = response.get(i + 1).toString();
                            result.put(field, value);
                        }
                        return result;
                    } else {
                        return new JsonObject();
                    }
                })
                .onFailure(throwable -> LOGGER.error("Failed to HGETALL key '{}': {}", key, throwable.getMessage(), throwable));
    }

    /**
     * 删除哈希表中一个或多个字段。
     *
     * @param key    哈希表的键
     * @param fields 要删除的字段列表
     * @return Future<Long> 成功删除的字段数量
     */
    public static Future<Long> hdel(String key, List<String> fields) {
        return getRedisAPI()
                .compose(api -> api.hdel(new ArrayList<>() {{
                    add(key);
                    addAll(fields);
                }}))
                .map(Response::toLong)
                .onFailure(throwable -> LOGGER.error("Failed to HDEL key '{}', fields '{}': {}", key, fields, throwable.getMessage(), throwable));
    }


    /**
     * 获取底层的 RedisAPI 实例。
     * 仅当您需要执行此工具类未直接封装的更复杂 Redis 命令时才使用。
     *
     * @return RedisAPI 实例的 Future
     */
    public static Future<RedisAPI> getRawRedisAPI() {
        return getRedisAPI();
    }
}